/**
 * @Author: liaoPengLiang
 * @File:  receiver
 * @Version: 1.0.0
 * @Date: 2023/3/24
 * @Description:
 */

package tunnelx

import (
	utils "github.com/alibaba/MongoShake/v2/common"
)

func Startup(cfg *Configuration) {
	factory := ReaderFactory{Name: cfg.Tunnel}
	reader := factory.Create(cfg.TunnelAddress)
	if reader == nil {
		return
	}

	repList := make([]*Replayer, cfg.ReplayerNum)
	for i := range repList {
		repList[i] = NewReplayer(i)
	}
	if err := reader.Link(repList); err != nil {
		return
	}
}

type Reader interface {
	/**
	 * Bridge of tunnel reader and aggregater(replayer)
	 *
	 */
	Link(aggregate []*Replayer) error
}

type ReaderFactory struct {
	Name string
}

func (factory *ReaderFactory) Create(address string) Reader {
	switch factory.Name {
	case utils.VarTunnelKafka:
		//return &KafkaReader{address: address}
	case utils.VarTunnelTcp:
		//return &TCPReader{listenAddress: address}
	case utils.VarTunnelRpc:
		return &RPCReader{address: address}
	case utils.VarTunnelMock:
		//return &MockReader{}
	case utils.VarTunnelFile:
		//return &FileReader{File: address}
	case utils.VarTunnelDirect:
		return nil
	default:
		return nil
	}
	return nil
}
